package utils;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class Kafka {
    // Kafka 服务器地址/主题名称
    public static void createTopic(String bootstrapServers,String topicName){
        // 分区数
        int numPartitions = 3;

        // 复制因子
        short replicationFactor = 1;

        // 配置Kafka AdminClient
        Properties adminClientProperties = new Properties();
        adminClientProperties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

        try (AdminClient adminClient = AdminClient.create(adminClientProperties)) {
            // 创建一个NewTopic对象
            NewTopic newTopic = new NewTopic(topicName, numPartitions, replicationFactor);

            // 创建主题
            adminClient.createTopics(Collections.singletonList(newTopic)).all().get();

            System.out.println("Kafka主题创建成功: " + topicName);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
}
